#include "SocketPool.h"

pthread_mutex_t SocketPool::_mutex_init = PTHREAD_MUTEX_INITIALIZER;
SocketPool *SocketPool::_instance = NULL;

SocketPool::SocketPool(Terminal terminal, int32 threadCnt, int32 retryTime)
{
    _terminal     = terminal;
    _threadCnt    = threadCnt;
    _retryTime    = retryTime;
    _threadSocket = new ThreadSocket[_threadCnt];
    for(int32 i=0; i<_threadCnt; i++)
    {
        _threadSocket[i].groupSocket = new GroupSocket[GetGroupCnt()];
        for(int32 j=0; j<GetGroupCnt(); j++)
        {
            _threadSocket[i].groupSocket[j].sCnt         = GetServerCnt(j);
            _threadSocket[i].groupSocket[j].serverSocket = new ServerSocket[GetServerCnt(j)];
            for(int32 k=0; k<GetServerCnt(j); k++)
            {
                _threadSocket[i].groupSocket[j].serverSocket[k].fd         = -1;
                _threadSocket[i].groupSocket[j].serverSocket[k].avail      = 1;
                _threadSocket[i].groupSocket[j].serverSocket[k].failedTime = GetNowTimeVal();
            }
        }
    }
}

SocketPool::~SocketPool()
{
    for(int32 i=0; i<_threadCnt; i++)
    {
        for(int32 j=0; j<GetGroupCnt(); j++)
        {
            delete [] _threadSocket[i].groupSocket[j].serverSocket;
            _threadSocket[i].groupSocket[j].serverSocket = NULL;
        }
        delete [] _threadSocket[i].groupSocket;
        _threadSocket[i].groupSocket = NULL;
    }
    delete [] _threadSocket;
    _threadSocket = NULL;
}

SocketPool *SocketPool::Instance()
{
    if(ConfManager::Instance()->GetIsDispatcher() <= 0)
    {
        WriteLog(NOTICE, "Init SocketPool error, the value of IsDispatcher is [%d].", ConfManager::Instance()->GetIsDispatcher());
        exit(-1);
    }

    if(_instance == NULL)
    {
        pthread_mutex_lock(&_mutex_init);

        if(_instance == NULL)
        {
            _instance = new SocketPool(ConfManager::Instance()->GetTerminal(), ConfManager::Instance()->GetWorkThreadCnt(), ConfManager::Instance()->GetRetryTime());
        }

        pthread_mutex_unlock(&_mutex_init);
    }

    return _instance;
}

void SocketPool::Start()
{
    WriteLog(NOTICE, "SocketPool is starting......");
    std::string ip;
    int32 port;
    int32 fd;
    for(int32 i=0; i<GetGroupCnt(); i++)
    {
        for(int32 j=0; j<GetServerCnt(i); j++)
        {
            for(int32 k=0; k<_threadCnt; k++)
            {
                _threadSocket[k].groupSocket[i].serverSocket[j].avail = false;
                _threadSocket[k].groupSocket[i].serverSocket[j].fd = -1;
                _threadSocket[k].groupSocket[i].serverSocket[j].failedTime = GetNowTimeVal();

                if(!GetIPPort(i, j, ip, port))
                {
                    WriteLog(WARNING, "Thread num[%d] get ip/port of group[%d] server[%d] error.", k, i, j);

                }
                else if((fd = GetSocketFd(ip, port)) < 0)
                {
                    WriteLog(WARNING, "Thread num[%d] connect [%s:%d] of group[%d] server[%d] error.", k, ip.c_str(), port, i, j);
                }
                else
                {
                    _threadSocket[k].groupSocket[i].serverSocket[j].avail = true;
                    _threadSocket[k].groupSocket[i].serverSocket[j].fd    = fd;
                    WriteLog(NOTICE, "Thread num[%d] connect [%s:%d] of group[%d] server[%d] successfully.", k, ip.c_str(), port, i, j);
                }
            }
        }
    }
    srand((int)time(0));
    WriteLog(NOTICE, "SocketPool start successfully.");
}

void SocketPool::Stop()
{
    WriteLog(NOTICE, "SocketPool is stopping......");
    for(int32 i=0; i<GetGroupCnt(); i++)
    {
        for(int32 j=0; j<GetServerCnt(i); j++)
        {
            for(int32 k=0; k<_threadCnt; k++)
            {
                _threadSocket[k].groupSocket[i].serverSocket[j].avail = false;
                close(_threadSocket[k].groupSocket[i].serverSocket[j].fd);
                WriteLog(NOTICE, "Thread num[%d] close fd[%d] of group[%d] server[%d] successfully.", k, _threadSocket[k].groupSocket[i].serverSocket[j].fd, i, j);
            }
        }
    }
    WriteLog(NOTICE, "SocketPool stop successfully.");
}

int32 SocketPool::GetGroupCnt()
{
    return _terminal.gCnt;
}

int32 SocketPool::GetServerCnt(int32 gNum)
{
    int32 gCnt = GetGroupCnt();
    if(gNum >= gCnt)
    {
        return 0;
    }

    return _terminal.group[gNum].sCnt;
}

bool SocketPool::GetIPPort(int32 gNum, int32 sNum, std::string &ip, int32 &port)
{
    int32 sCnt = GetServerCnt(gNum);
    if(sNum >= sCnt)
    {
        return false;
    }

    ip   = _terminal.group[gNum].server[sNum].ip;
    port = _terminal.group[gNum].server[sNum].port;

    return true;
}

int32 SocketPool::GetSocketFd(std::string IP, int32 port)
{
    int socketFd = -1;

    if((socketFd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
    {
        return -1;
    }

    struct sockaddr_in servaddr;
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(port);
    inet_pton(AF_INET, IP.c_str(), &servaddr.sin_addr);

    if(connect(socketFd, (struct sockaddr*)&servaddr, sizeof(servaddr)) < 0)
    {
        close(socketFd);
        return -1;
    }

    return socketFd;
}

bool SocketPool::GetSocket(int32 threadNum, int32 groupNum, int32 &fd, int32 &sNo)
{
    int32 sCnt = GetServerCnt(groupNum);
    if(sCnt <= 0 || threadNum >= _threadCnt)
    {
        return false;
    }

    sNo = rand()%(sCnt);
    int32 start = sNo;
    do{
        if(_threadSocket[threadNum].groupSocket[groupNum].serverSocket[sNo].avail == true)
        {
            fd = _threadSocket[threadNum].groupSocket[groupNum].serverSocket[sNo].fd;
            return true;
        }
        else
        {
            std::string ip;
            int32 port;
            int32 fd;

            timeval_t now  = GetNowTimeVal();
            timeval_t fail = _threadSocket[threadNum].groupSocket[groupNum].serverSocket[sNo].failedTime;
            if(TimeDiff(fail, now) > (double)_retryTime)
            {
                GetIPPort(groupNum, sNo, ip, port);
                if((fd = GetSocketFd(ip, port)) < 0)
                {
                    WriteLog(WARNING, "Thread num[%d] reconnect [%s:%d] of group[%d] server[%d] error.", threadNum, ip.c_str(), port, groupNum, sNo);
                }
                else
                {
                    _threadSocket[threadNum].groupSocket[groupNum].serverSocket[sNo].avail = true;
                    _threadSocket[threadNum].groupSocket[groupNum].serverSocket[sNo].fd    = fd;
                    WriteLog(NOTICE, "Thread num[%d] reconnect [%s:%d] of group[%d] server[%d] successfully.", threadNum, ip.c_str(), port, groupNum, sNo);
                    return true;
                }
            }
        }
        sNo = ((++sNo)%sCnt);
    }while(sNo != start);

    return false;
}

void SocketPool::ResetSocket(int32 threadNum, int32 groupNum, int32 sNo)
{
    _threadSocket[threadNum].groupSocket[groupNum].serverSocket[sNo].avail      = false;
    _threadSocket[threadNum].groupSocket[groupNum].serverSocket[sNo].fd         = -1;
    _threadSocket[threadNum].groupSocket[groupNum].serverSocket[sNo].failedTime = GetNowTimeVal();
}
